package com.systar.activity.process.online;

import java.nio.ByteBuffer;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;

import com.systar.activity.ZookeeperConnector;
import com.systar.activity.events.IActivityEvent;
import com.systar.activity.monitor.ActivityManager;
import com.systar.activity.monitor.ActivitySource;
import com.systar.activity.process.ActivityStack;

/**
 * This class is in charge to send data to Zookeeper server
 */
public class ZookeeperNotifier extends ZookeeperConnector implements Runnable
{
	private final ActivityManager manager;
	private final ActivityStack eventStack;
	private ScheduledExecutorService scheduler;

	public ZookeeperNotifier(ActivityManager manager, String zkHostPort)
	{
		super(zkHostPort);
		
		this.manager = manager;
		eventStack = manager.createEventDelegation();
	}

	@Override
	protected void startupInit() throws Exception
	{
		scheduler = Executors.newSingleThreadScheduledExecutor();

		// send source declarations
		for (ActivitySource source : manager.getSources())
		{
			String path = "/" + source.getName();
			Stat stat = getZooKeeper().exists(path, false);
			if (stat == null)
			{
				getZooKeeper().create(path, new byte[]{0,0,0,0,0}, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
		}
		// run pump thread
		scheduler.scheduleWithFixedDelay(this, 0, 5, TimeUnit.SECONDS);
	}

	@Override
	public void deactivate()
	{
		if (scheduler != null)
			scheduler.shutdown();
		super.deactivate();
	}
	
	@Override
	public void run()
	{
		// post activity events
		for (IActivityEvent event : eventStack.get())
		{
			ActivitySource source = event.getSource();
			String path = "/" + source.getName();
			try
			{
				ByteBuffer data = ByteBuffer.wrap(getZooKeeper().getData(path, false, null));
				event.marshal(data);
				getZooKeeper().setData(path, data.array(), -1);
			}
			catch (KeeperException e1)
			{
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
			catch (InterruptedException e1)
			{
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
		}
	}

}
